package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.CoursePaperBean;
import com.atguigu.edu.realtime.bean.InteractionCourseCommentBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;

//卢顺祥  课程评价次数和总评价分数聚合表
public class DwsInteractionCourseCommentWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)
        // TODO 3. 从 Kafka dwd_course_test_exam 主题读取课程数据
        String topic = "dwd_interaction_course_comment";
        String groupId = "dws_trade_province_order_window";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> source = env.addSource(kafkaConsumer);

        // TODO 4.转换数据结构
        SingleOutputStreamOperator<JSONObject> mappedStream = source.map(JSON::parseObject);
        //{"course_id":"62","create_time":"2022-10-21 13:04:52",
        // "user_id":"23","id":"31946","date_id":"2022-10-21","review_stars":"5","ts":"1666328692"}


        // TODO 7. 转换数据结构
        SingleOutputStreamOperator<InteractionCourseCommentBean> javaBeanStream = mappedStream.map(
                jsonObj -> {
                    String course_id = jsonObj.getString("course_id");
                    Double review_stars = jsonObj.getDouble("review_stars");
                    Long ts = jsonObj.getLong("ts") * 1000L;

                    InteractionCourseCommentBean tradeProvinceOrderWindow = InteractionCourseCommentBean.builder()
                            .course_id(course_id)
                            .orderCount(1L)
                            .orderIdSet(new HashSet<String>(
                                    Collections.singleton(course_id)
                            ))
                            .review_stars(review_stars)
                            .ts(ts)
                            .build();
                    return tradeProvinceOrderWindow;
                }
        );
        //InteractionCourseCommentBean(stt=null, edt=null, course_id=427, orderCount=1, orderIdSet=[427], review_stars=5.0, ts=1666332564000)
        //javaBeanStream.print();


        // TODO 8. 设置水位线
        SingleOutputStreamOperator<InteractionCourseCommentBean> withWatermarkStream = javaBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<InteractionCourseCommentBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<InteractionCourseCommentBean>() {
                                    @Override
                                    public long extractTimestamp(InteractionCourseCommentBean javaBean, long recordTimestamp) {
                                        return javaBean.getTs();
                                    }
                                }
                        )
        );

        // TODO 9. 按照试卷 ID 分组
        KeyedStream<InteractionCourseCommentBean, String> keyedByProIdStream =
                withWatermarkStream.keyBy(InteractionCourseCommentBean::getCourse_id);

        // TODO 10. 开窗
        WindowedStream<InteractionCourseCommentBean, String, TimeWindow> windowDS = keyedByProIdStream.window(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)
        ));

        // TODO 11. 聚合计算
        SingleOutputStreamOperator<InteractionCourseCommentBean> reducedStream = windowDS.reduce(
                new ReduceFunction<InteractionCourseCommentBean>() {
                    @Override
                    public InteractionCourseCommentBean reduce(InteractionCourseCommentBean value1, InteractionCourseCommentBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(
                                value2.getOrderIdSet()
                        );
                        value1.setReview_stars(
                                value1.getReview_stars() + value2.getReview_stars()
                        );
                        value1.setOrderCount(value1.getOrderCount()+value2.getOrderCount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<InteractionCourseCommentBean, InteractionCourseCommentBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<InteractionCourseCommentBean> elements, Collector<InteractionCourseCommentBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        for (InteractionCourseCommentBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            //element.setOrderCount((long) element.getOrderIdSet().size());
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );
        SinkFunction<InteractionCourseCommentBean> jdbcSink = MyClickhouseUtil.<InteractionCourseCommentBean>getSinkFunction(
                "insert into Dws_Interaction_Course_Comment_Window values(?,?,?,?,?,?)"
        );
        reducedStream.<InteractionCourseCommentBean>addSink(jdbcSink);


        env.execute();

    }
}
